package com.chb.flink.source

import java.text.SimpleDateFormat
import java.util.Date

import org.apache.flink.api.scala.ExecutionEnvironment


object CollectionSource {
    def main(args: Array[String]): Unit = {
        //初始化Flink的Streaming（流计算）上下文执行环境
        val env = ExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
        //导入隐式转换
        import org.apache.flink.api.scala._

        //读取数据
        var data = env.fromCollection(Array(
            new StationLog("001", "186", "189", "busy", 1577071519462L, 0),
            new StationLog("002", "186", "188", "busy", 1577071520462L, 0),
            new StationLog("003", "183", "188", "busy", 1577071521462L, 0),
            new StationLog("004", "186", "188", "success", 1577071522462L, 32)
        ))

        data.print()
    }
}

/*
    * 通信基站日志数据
    * @param sid 基站ID
    * @param callOut 主叫号码
    * @param callIn 被叫号码
    * @param callType 通话类型eg:呼叫失败(fail)，占线(busy),拒接（barring），接通(success)
    * @param callTime 呼叫时间戳，精确到毫秒
    * @Param duration 通话时长 单位：秒
*/
case class StationLog(sid: String, var callOut: String, var callIn: String, callType: String, callTime: Long, duration: Long) {
    val df = new SimpleDateFormat("HH:mm:ss")

    override def toString: String = "StationLog[" + sid + "," + callOut + "," + callIn + "," + callType + "," + callTime + "," +duration + "," + df.format(new Date(callTime)) + "]"
}